#! /usr/bin/env python

"""
This is simple file server the implements a flat file service that allows
clients to to put and get file data into its underlying file store.

The file store consist of a standard file sytems directory where data
(both files and directories) is stored as a series of actual files.
Files are given a unique file id via a call to uuid1 which generates
a globally unique number according to RFC 4122. Next they are assigned
a storage location based on the next available sequence number according
to the index file. This index file consists of fid,seqno pairs telling
the system the seqnos in use by existing files.

The allocated sequence number is mapped onto a physical file location
using the following logic:

1. seqno = get current seqno for fid
2. if seqno not assigned reserve new one (get next available seqno)
2. str = convert sequence number to 8 digit hex string, index=[0..7]
3. fname = data-dir / str[0..2] / str[3..4] / str[5..6] / fid

So for example:

1. fid   = b31c8cc6-2999-11e1-8f97-292cfd9858c0
2. seqno = 2
3  seqstr = 00000002
3. fname = data-dir/00/00/00/b31c8cc6-2999-11e1-8f97-292cfd9858c0.data

We use this layout to prevent thousands of files being stored in the
one directory. We store a files attributes or meta-data using the
extended file attributes feature of modern unix file systems.
We use the user namespace and the following tags:

 user.dfs_ftype - file type
 user.dfs_ufid  - file id

Note that it is the client proxy that does the job of mapping directory names
to file ids. Also the file server is stateless in that there is really
no concept of an open file although a callback operation is provided that
allows a client to be informed of any file modifcation such as update or
deletes by other clients.

To start the server use:

  python file_server.py hostname port data-directory

Clients can access the file server using the FileSocket class which
implements the client side of a rpc like interface using tcp sockets.

We support these operations:

1) create - create a new file and return its file id
2) remove - delete the file given its file id
3) stat - get a files attributes
4) get - read a section of a file 
5) put - update a section of a file
6) register client callback for file modifcations
7) unregister client callback

Sample usage:
============

def test():
    addr = (localhost,1000)
    conn = FileSocket(addr)
    fattr = FTYPE_FILE
    fid  = conn.create(fattr)

    str1 = 'some data'
    conn.put(fid, 0, str1)
    str2 = conn.get(fid, 0, len(str1))

    conn.close()

    assert str1 == str2


Dependencies:
    python-xattr - Python wrapper for extended filesystem attributes
      Fast install: yum install python-xattr
      Slow install: download pyxattr-0.5.2.tar.gz & run python setup.py
"""


import socket,threading,string
import marshal,os,errno
import uuid,xattr
import StringIO

from dfs_common import *

# extended attribute value where we store our metadata
XATTR_DFS_UFID = 'user.dfs_ufid'
XATTR_DFS_FTYPE = 'user.dfs_ftype'

file_watchers = {}

def add_watcher(fid, watcher):
    "Add a client to a files watch list."
    try:
        watch_list = file_watchers[fid]
    except KeyError:
        watch_list = []
        file_watchers[fid] = watch_list
    watch_list.append(watcher)

def inform_watchers(sender, fid, event):
    "Tell all client except the sender about a file event."
    try:
        watch_list = file_watchers[fid]
        for watcher in watch_list:
            if watcher != sender and not watcher.closed():
                print 'sending event', event
                watcher.send_event(event)
    except KeyError:
        pass

def remove_watcher(fid, watcher):
    "Remove a client from a files watch list."
    try:
        watch_list = file_watchers[fid]
        watch_list.remove(watcher)
    except KeyError,ValueError:
        pass

def reset_watchers(fid):
    "Remove all clients from a files watch list."
    try:
        del file_watchers[fid]
    except KeyError,ValueError:
        pass


class FileIndexError(DFSError):
    "Raised if a invalid entry is found."

    def __init__(self, idx, reason):
        self.idx = idx
        self.reason = reason

    def __str__(self):
        return 'line %d: %s' %(self.idx,  self.reason)

class MissingEntry(DFSError):
    "Raised  a when an entry is not found."
    pass

class FileIndex:
    """This implements a thread safe file index that allocates seqno or slot
    postions for a given file identifier (fid). The index file is
    a simple text file with a fid,seqno pair on each line
    """
    def __init__(self, fpath):

        self.lock = threading.Lock()
        self.fname = '%s/%s' %( fpath, 'files.idx' )
        self.fid_dict = None
        self.used_seqno = None
        # this may take a while
        self.load_entries()
        self.value = self.next_free(0)

    # load all numbers stored in a file into set
    def load_entries(self):
        seqno_set = set()
        fid_dict = {}
        if os.path.exists(self.fname):
            with open(self.fname) as f:
                for i,line in enumerate(f):
                    line = line.strip()
                    if len(line) == 0 or line.startswith('#'):
                        # skip blank or comment lines
                        continue
                    fields = line.split(' ', 2)
                    if len(fields) != 2:
                        raise FileIndexError(i, 'Expected 2 fields')
                    # verify fid
                    field = fields[0]
                    if len(field) == 0:
                        raise FileIndexError(i, 'fid len 0 ' + field)
                    if field in fid_dict:
                        raise FileIndexError(i, 'Duplicate fid ' + field)
                    fid = field
                    # verify seqno
                    seqno = str2num(fields[1])
                    if seqno is None:
                        raise FileIndexError(i, 'seqno not a valid number ' + field)
                    if seqno in seqno_set:
                        raise FileIndexError(i, 'Duplicate seqno' + field)
                    # update 
                    fid_dict[fid] = seqno
                    seqno_set.add(seqno)
        self.fid_dict = fid_dict
        self.used_seqno = seqno_set

    # calculate the next free sequnce number
    def next_free(self, seqno):
        while seqno in self.used_seqno:
            seqno = seqno + 1
        return seqno

    def gen_entry(self, fid, seqno):
        return '%s %d\n' %  (fid, seqno)

    # save all numbers in a set to a file
    def add_entry(self, fid, seqno):
        with open(self.fname,'a') as f:
            f.write(self.gen_entry(fid, seqno))
        self.fid_dict[fid] = seqno
        self.used_seqno.add(seqno)

    # TODO replace with fixed length entries
    def del_entry(self, fid, seqno):
        buf = StringIO.StringIO()
        del self.fid_dict[fid]
        self.used_seqno.remove(seqno)
        for fid, seqno in self.fid_dict.iteritems():
            buf.write(self.gen_entry(fid,seqno))
        with open(self.fname,'w') as f:
            f.write(buf.getvalue())

    def get_seqno(self, fid):
        return self.fid_dict.get(fid, None)

    def free(self, fid):
        self.lock.acquire()
        try:
            seqno = self.get_seqno(fid)
            if seqno is not None:
                self.del_entry(fid, seqno)
        finally:
            self.lock.release()

    # return the next available sequence number
    # we use a lock to prevent concurrent updates
    def reserve(self, fid):
        self.lock.acquire()
        try:
            seqno = self.next_free(self.value)
            self.add_entry(fid, seqno)
            self.value = seqno
        finally:
            self.lock.release()
        return seqno


class ServerSocket(MessageSocket):
    """This class implements the rpc server side of the file server.
    One is created per client socket connection running in a seperate thread.
    """
    CREATE_FLAGS = os.O_CREAT + os.O_WRONLY + os.O_TRUNC
        
    def __init__(self, sock, addr, root_dir, index):
        MessageSocket.__init__(self, sock, addr)
        
        self.root_dir = root_dir
        self.index = index

        print 'Client', self.addr, 'connected'

    def handle_closed(self):
        print 'Client', self.addr, 'disconnected'

    # convert fid to file path
    def gen_filename(self, fid, reserve=False):
        seqno = self.index.get_seqno(fid)
        if seqno is None:
            if not reserve:
                raise OSError(errno.ENOENT, 'No seqno', fid)
            seqno = self.index.reserve(fid)
        astr ='%08X' % seqno
        assert len(astr) == 8
        fpath = '%s/%s/%s/%s/%s.data' %( self.root_dir, astr[0:2], astr[2:4], astr[4:6], fid )
        return fpath

    def next_fid(self):
        return str(uuid.uuid1())

    # get file attrbutes
    def get_fattr(self, src):
        fattr = FileAttr()
        if type(src) is str:
            # we have a file name
            fstat = os.stat(src)
            fattr.raw_path = src
        else:
            # we have a file descriptor
            fstat = os.fstat(src)
        fattr.size = fstat.st_size
        fattr.mtime = fstat.st_mtime
        fattr.fid = xattr.get(src, XATTR_DFS_UFID)
        fattr.ftype = int(xattr.get(src, XATTR_DFS_FTYPE))
        return fattr

    def get_addr(self):
        return '%s:%d' %( self.addr[0], self.addr[1])

    # add client socket to a files watch list
    def register(self, fid):
        add_watcher(fid, self)
        return self.stat(fid)

    # remove client socket from a files watch list
    def unregister(self, fid):
        remove_watcher(fid, self)
        return (0,'')

    # Create a files storage lcoation (i.e mkdir -p)
    def create_path(self, path):
        print 'Path', path
        dname,fname = path.rsplit('/',1)
        if not os.path.exists(dname):
            os.makedirs(dname)

    # create a new file in the file store with given id & type
    def create_wfid(self, fid, ftype):
        try:
            fpath = self.gen_filename(fid, True)
            print 'Creating', fpath, fid
            self.create_path(fpath)
            fd = os.open(fpath, self.CREATE_FLAGS)
            os.close(fd)
            xattr.set(fpath, XATTR_DFS_UFID, fid)
            xattr.set(fpath, XATTR_DFS_FTYPE, str(ftype))
            fattr = self.get_fattr(fpath)
        except OSError, e:
            return (e.errno, 0)
        return (0, (fid, fattr))

    def create(self, ftype):
        return self.create_wfid(self.next_fid(), ftype)

    # remove a file from the file store
    def remove(self, fid):
        try:
            fpath = self.gen_filename(fid)
            os.remove(fpath)
            self.index.free(fid)
            inform_watchers(self, fid, ('deleted',(fid)))
            reset_watchers(fid)
        except OSError, e:
            return (e.errno, '')
        return (0,'')

    # return a files attributes
    def stat(self, fid):
        try:
            fpath = self.gen_filename(fid)
            fattr = self.get_fattr(fpath)
        except OSError, e:
            return (e.errno, '')
        return (0,fattr)

    # return a list of stats (useful for clients do lsdir)
    def stat_list(self, fid_list):
        try:
            fattr_list = []
            for fid in fid_list:
                fpath = self.gen_filename(fid)
                fattr = self.get_fattr(fpath)
                fattr_list.append(fattr)
        except OSError, e:
            return (e.errno, '')
        return (0,fattr_list)
   
    # read size bytes from file starting at offset pos
    # return file attribute + read bytes
    def get(self, fid, pos, size):
        try:
            fpath = self.gen_filename(fid)
            fd = os.open(fpath, os.O_RDONLY)
            os.lseek(fd, pos, os.SEEK_SET)
            data_len = min(DFS_BUFSIZE, size)
            data = os.read(fd, data_len)
            fattr = self.get_fattr(fd)
            os.close(fd)
        except OSError, e:
            return (e.errno, '')
        return (0, (fattr, data))

    # write given data to file starting at offset pos
    # return file attrbute
    def put(self, fid, pos, data):
        try:
            fpath = self.gen_filename(fid)
            fd = os.open(fpath, os.O_WRONLY)
            os.lseek(fd, pos, os.SEEK_SET)
            prev_fattr = self.get_fattr(fd)
            total_write = 0
            while total_write < len(data):
                nwrite = os.write(fd, data[total_write:])
                total_write += nwrite
            fattr = self.get_fattr(fd)
            os.close(fd)
            # tell all proxys that file has changed
            inform_watchers(self, fid, ('modified',(fid, fattr)))
        except OSError, e:
            return (e.errno, 0)
        return (0, (fattr,prev_fattr))

class FileSocket(MessageSocket):
    """This class implements the rpc client side of the file server
    see server class above for description of operations.
    """
    def __init__(self, addr):

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect(addr)
        MessageSocket.__init__(self, sock, addr)

    def register(self, fid):
        return self.call('register', fid)

    def unregister(self, fid):
        return self.call('unregister', fid)

    def create_wfid(self, fid, fattr):
        return self.call('create_wfid', fid, fattr)

    def create(self, fattr):
        return self.call('create', fattr)
    
    def remove(self, fid):
        return self.call('remove', fid)

    def stat(self, fid):
        return self.call('stat', fid)

    def stat_list(self, fid_list):
        return self.call('stat_list', fid_list)

    def get(self, fid, pos, size):
        return self.call('get', fid, pos, size)

    def put(self, fid, pos, data):
        return self.call('put', fid, pos, data)


if __name__ == '__main__':
    import sys,socket
    USAGE = "Usage: server.py addr port rdir"
    if len(sys.argv) != 4:
        print USAGE
        sys.exit(1)

    saddr = (sys.argv[1], int(sys.argv[2]))
    data_dir = sys.argv[3]

    if not os.path.isdir(data_dir):
        print 'Root directory not found:', data_dir
        sys.exit(2)

    index = FileIndex(data_dir)

    ss = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    ss.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,1)

    ss.bind(saddr)
    ss.listen(128)

    print 'Listening on', saddr

    run = True

    while run:
        try:
            cs,caddr = ss.accept()
            client = ServerSocket(cs, caddr, data_dir, index)
            client.start()
        except KeyboardInterrupt:
            run = False

    sys.exit(0)
